package com.sojson.util.mq.redis.interfaces.impl.redis;

import java.io.IOException;

import javax.annotation.PostConstruct;

import com.sojson.util.StringUtil;
import com.sojson.util.cache.redis.impl.RedisUtilImpl;
import com.sojson.util.log.LogUtil;
import com.sojson.util.mq.redis.bean.MqHandler;

import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

/**
 * Redis消息工具类实现类
 *
 * @author liu
 * @date 2022-01-20
 */
@Slf4j
@SuppressWarnings("synthetic-access")
public class RedisMqUtilImpl {

    private static RedisMqUtilImpl redisMqUtilImpl;
    private RedisUtilImpl redis = RedisUtilImpl.getInstances();

    /**
     * 广播消息
     *
     * @param channel   通道
     * @param msg       消息
     * @return
     */
    public Long broadcast(String channel, String msg) {
        Jedis jds = null;
        try {
            jds = redis.getJedis();
            Long publish = jds.publish(channel, msg);
            log.info("Redis消息: 发送{}消息{}", channel, msg);
            return publish;
        } finally {
            try {
                redis.close(jds);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    /**
     * 订阅消息
     *
     * @param channel   通道
     * @param handler   处理实现
     * @return
     */
    public void subscribe(String channel, MqHandler handler) {
        Jedis jds = null;
        try {
            LogUtil.setTraceId("Redis消息");

            jds = redis.getJedis();
            jds.subscribe(new JedisPubSub() {
                @Override
                public void onMessage(String channel, String msg) {
                    log.info("收到{}消息{}", channel, msg);
                    handler.run(msg);
                }

                @Override
                public void onSubscribe(String channel, int subscribedChannels) {
                    log.info("订阅消息{}", channel);
                }

                @Override
                public void onUnsubscribe(String channel, int subscribedChannels) { // 取消订阅 会调用
                    log.info("取消订阅{}", channel);
                }
            }, channel);
        } finally {
            try {
                redis.close(jds);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
            LogUtil.clearTraceId();
        }

    }

    public static RedisMqUtilImpl getInstances() {
        if (StringUtil.isBlankObject(redisMqUtilImpl)) {
            redisMqUtilImpl = new RedisMqUtilImpl();
        }
        return redisMqUtilImpl;
    }

    /**
     * 创建消费者
     *
     * @throws IOException
     */
    @PostConstruct
    public void init() throws IOException {
        RedisMqUtilImpl redis = RedisMqUtilImpl.getInstances();
        redis.subscribe("test", new MqHandler() {
            @Override
            public void run(String msg) {
                System.out.println("+++++++++++++++++++++++");
                System.out.println(msg);
                System.out.println("+++++++++++++++++++++++");
            }
        });
    }

    public static void main(String[] args) {
        // 发送消息
        RedisMqUtilImpl redis = RedisMqUtilImpl.getInstances();
        redis.broadcast("test", "测试");
    }

}